package com.lty.netty5.client;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * 多连接客户端
 * @author -琴兽-
 *
 */
public class MultClientPool implements Runnable{
	/**
	 * 线程池
	 */
	private Executor executor = Executors.newCachedThreadPool();
	
	/**
	 * 服务类
	 */
	private Bootstrap bootstrap = new Bootstrap();
	
	/**
	 * 会话
	 */
	private List<Channel> channels = new ArrayList<>();
	
	/**
	 * 引用计数
	 */
	private final AtomicInteger index = new AtomicInteger();
	
	/**
	 * 任务队列
	 */
	private final Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<Runnable>();
	
	/**
	 * 初始化
	 * @param count
	 */
	public void init(int count){
		
		//worker
		EventLoopGroup worker = new NioEventLoopGroup();
		
		//设置线程池
		bootstrap.group(worker);
		
		//设置socket工厂
		bootstrap.channel(NioSocketChannel.class);
		
		//设置管道
		bootstrap.handler(new ChannelInitializer<Channel>() {

			@Override
			protected void initChannel(Channel ch) throws Exception {
				ch.pipeline().addLast(new StringDecoder());
				ch.pipeline().addLast(new StringEncoder());
				ch.pipeline().addLast(new ClientHandler());
			}
		});
		
		for(int i=1; i<=count; i++){
			ChannelFuture future = bootstrap.connect("127.0.0.1", 10102);
			channels.add(future.channel());
		}
	}
	
	/**
	 * 获取会话
	 * @return
	 */
	public Channel nextChannel(){
		return getFirstActiveChannel(0);
	}
	
	private Channel getFirstActiveChannel(int count){
		final Channel channel = channels.get(Math.abs(index.getAndIncrement() % channels.size()));
		if(!channel.isActive()){
			//重连
			addTask(new Runnable() {
				//重连任务
				@Override
				public void run() {
					if(channels.indexOf(channel) == -1){
						return ;
					}
					
					Channel newChannel = bootstrap.connect("127.0.0.1", 10102).channel();
					channels.set(channels.indexOf(channel), newChannel);
				}
			});
			if(count >= channels.size()){
				throw new RuntimeException("no can use channel");
			}
			executor.execute(this);
			return getFirstActiveChannel(count + 1);
		}
		return channel;
	}
	
	/**
	 * 注册一个任务并激活selector
	 * 
	 * @param task
	 */
	protected final void addTask(Runnable task) {
		taskQueue.add(task);
	}

	/* 
	* <p>Title: run</p> 
	* <p>Description: </p>  
	* @see java.lang.Runnable#run() 
	*/
	@Override
	public void run() {
		processTaskQueue();
	}
	
	/**
	 * 执行队列里的任务
	 */
	private void processTaskQueue() {
		for (;;) {
			final Runnable task = taskQueue.poll();
			if (task == null) {
				break;
			}
			task.run();
		}
	}

}
